package com.braska.grave.netty.manager;

import com.braska.grave.netty.bean.invoker.NettyInterfaceInvoker;
import com.braska.grave.netty.sender.RequestSender;
import io.netty.channel.Channel;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author shenyuhang
 * @date 2020/3/31
 **/
public class NettyChannelManager extends ConcurrentHashMap<SocketAddress, Channel> {
    private CopyOnWriteArrayList<Channel> channels;
    private AtomicInteger roundRobin;
    private RequestSender sender;

    public NettyChannelManager(RequestSender sender) {
        this.channels = new CopyOnWriteArrayList<>();
        this.roundRobin = new AtomicInteger(0);
        this.sender = sender;
        NettyInterfaceInvoker.getInstance().setSender(this.sender);
    }

    @Override
    public synchronized Channel put(SocketAddress address, Channel channel) {
        channels.add(channel);
        return super.put(address, channel);
    }

    public synchronized void remove(Channel channel) {
        SocketAddress remotePeer = channel.remoteAddress();
        if (remotePeer != null) super.remove(remotePeer);
        channels.remove(channel);
    }

    public Channel take() {
        if (this.size() > 0) {
            int size = channels.size();
            int index = (roundRobin.getAndAdd(1) + size) % size;
            return channels.get(index);
        }
        return null;
    }

    public synchronized void refresh(List<String> addressList) {
        if (addressList == null || addressList.size() == 0) {
            for (final Channel channel : channels) {
                SocketAddress remotePeer = channel.remoteAddress();
                Channel handler_node = this.get(remotePeer);
                handler_node.close();
            }
            channels.clear();
            this.clear();
            return;
        }

        HashSet<SocketAddress> newAllServerNodeSet = new HashSet<>();
        for (int i = 0; i < addressList.size(); ++i) {
            String[] hostAndPort = addressList.get(i).split(":");
            if (hostAndPort.length == 2) {
                String host = hostAndPort[0];
                int port = Integer.parseInt(hostAndPort[1]);
                final SocketAddress remotePeer = new InetSocketAddress(host, port);
                newAllServerNodeSet.add(remotePeer);
            }
        }

        for (final SocketAddress serverNodeAddress : newAllServerNodeSet) {
            Channel channel = this.get(serverNodeAddress);
            if (channel != null && channel.isOpen()) {
                // logger.info("当前服务节点已存在,无需重新连接.{}",serverNodeAddress);
            } else {
                if (channel != null) {
                    this.remove(channel);
                }
                connectServerNode(serverNodeAddress);
            }
        }
        for (int i = 0; i < channels.size(); ++i) {
            Channel channel = channels.get(i);
            SocketAddress remotePeer = channel.remoteAddress();
            if (remotePeer != null && !newAllServerNodeSet.contains(remotePeer)) {
                // logger.info("删除失效服务节点 " + remotePeer);
                Channel channel_node = this.get(remotePeer);
                if (channel_node != null) {
                    channel_node.close();
                }
                channels.remove(channel);
                this.remove(remotePeer);
            }
        }
    }

    private void connectServerNode(SocketAddress address) {
        try {
            Channel channel = sender.connect(address);
            this.put(address, channel);
        } catch (InterruptedException e) {
            e.printStackTrace();
            // logger.info("未能成功连接到服务器:{}",address);
        }
    }
}
